package com.atguigu.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author gmd
 * @desc 自定义Sink输出，通过富函数（RichSinkFunction）实现
 * @since 2024-11-26 11:15:00
 */
public class SinkCustom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<String> sensorDS = env.socketTextStream("127.0.0.1", 7777);

        sensorDS.addSink(new MySink2());
        env.execute();
    }

    public static class MySink extends RichSinkFunction<String> {
        Connection conn = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 在这里创建连接
            // conn = new xxxx
        }

        @Override
        public void close() throws Exception {
            super.close();
            // 做一些清理、销毁连接
        }

        /**
         * sink的核心逻辑，写出的逻辑就写在这个方法里
         *
         * @param value   数据
         * @param context 上下文
         * @throws Exception 异常
         */
        @Override
        public void invoke(String value, Context context) throws Exception {
            // 写出逻辑
            // 这个方法是 来一条数据，调用一次,所以不要在这里创建 连接对象
        }

    }

    /**
     * 自定义 Sink 类。这个类实现了一个简单的 Sink，将数据写入 MySQL 数据库。
     */
    public static class MySink2 extends RichSinkFunction<String> {

        /**
         * 数据库连接
         */
        private Connection conn = null;

        /**
         * 预编译 SQL 语句
         */
        private PreparedStatement preparedStatement = null;

        /**
         * 打开连接
         *
         * @param parameters 参数
         * @throws Exception 异常
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 在这里创建连接
            String url = "jdbc:mysql://localhost:3306/icv";
            String user = "root";
            String password = "123456";
            conn = DriverManager.getConnection(url, user, password);
            String sql = "INSERT INTO sensor_data(flinkData, dateTime) VALUES (?, ?)";
            preparedStatement = conn.prepareStatement(sql);
        }

        /**
         * 关闭连接
         *
         * @throws Exception 异常
         */
        @Override
        public void close() throws Exception {
            super.close();
            // 做一些清理、销毁连接
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (conn != null) {
                conn.close();
            }
        }

        /**
         * sink 的核心逻辑，写出的逻辑就写在这个方法里
         *
         * @param value   数据
         * @param context 上下文
         * @throws Exception 异常
         */
        @Override
        public void invoke(String value, Context context) throws Exception {
            // 写出逻辑
            // 这个方法是来一条数据，调用一次，所以不要在这里创建连接对象
            // 序号就是替换上面的?占位符
            preparedStatement.setString(1, value);
            preparedStatement.setDate(2, new Date(System.currentTimeMillis()));
            preparedStatement.executeUpdate();
        }

        /**
         * 数据表SQL
         * CREATE TABLE `sensor_data` (
         *   `flinkData` varchar(255) DEFAULT NULL,
         *   `dateTime` datetime DEFAULT NULL
         * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
         */
    }

}
